from celery import shared_task
from django.utils import timezone
from django.forms.models import model_to_dict
from django.db.models import Avg, Max, Min, Count, Sum
from fsygManager.PulsarSDK import SendAsyncMessage
from fsygManager.fsygSDK import requests_SDK
from iClouds.apidev import hash_md5, has_Reqsn_Callme
from CallManager import tasks as CALL_TASKS
from MoniterManager.tasks import doMoniterPulsarScreen
from PlaceManager.tasks import has_pid_from_idcode
from VipManager.tasks import UpDownEventFromFsygChange,UpDownEventFromFsygPay,UpDownEventFromFsygOnn,UpDownEventFromFsygOff

from .serializers import (
    FsygShopintoListSerializer,
    FsygShopreqsnListSerializer,
    AllinpayCallmeCreatSerializer,
)
from mysqldb import redis as REDIS
from mysqldb.models import *
from datetime import datetime, date, timedelta
import time, json
import random
import logging

logger = logging.getLogger(__name__)

from django.conf import settings


@shared_task
def CreateCallmeClearReqsn(pid, source):
    """创建清扫订单"""
    # 获取配置
    try:
        conf = AllinpayCallmeConf.objects.get(status=1, name="清扫服务", pid=pid)
    except BaseException as err:
        return {"rescode": 444, "resmsg": "不支持清扫服务"}
    # 创建记录
    r = AllinpayCallmeCreatSerializer(
        data={
            "pid": pid,
            "source": source,
            "rtype": conf.id,
            "status": 0,
            "current": datetime.now(),
        }
    )
    if r.is_valid():
        r.save()
        # 呼叫派单
        res = CALL_TASKS.Dispatch_Reqsn(**r.data)
        # logger.info(f"[fsyg] 创建清扫单成功")
        return {"rescode": 200, "resmsg": "创建清扫单成功"}
    else:
        logger.warn(f"[fsyg] 创建清扫单失败")
        return {"rescode": 444, "resmsg": "创建清扫单失败"}


@shared_task
def doRequestProgress(barcode, hash):
    """
    查询上机信息 2.1
    barcode,  string 场所编码
    hash=None, string 身份证MD5
    """
    API = requests_SDK()
    url = "/channel/bar/progress"
    params = {"wb_id": barcode, "id_card_hash": hash, "code": API.code}
    r = API.do_get(url, params=params)
    if not r:
        logger.warn(f"[user_info] [{hash}] 查询 上机信息 请求失败")
        return {"rescode": 400, "resmsg": "请求失败，无数据返回。"}
    else:
        return {"rescode": 200, "resmsg": r}


@shared_task
def doRequestUser(barcode, hash):
    """
    查询用户信息  2.2
    barcode,  string 场所编码
    hash=None, string 身份证MD5
    """
    # logger.info(f'[{barcode}] [] 收到异步请求 查询用户信息')
    API = requests_SDK()
    url = "/channel/bar/user"
    params = {"wb_id": barcode, "id_card_hash": hash, "code": API.code}
    r = API.do_get(url, params=params)
    if not r:
        logger.warn(f"[query_member] [{hash}] 查询 用户信息 请求失败")
        return {"rescode": 400, "resmsg": "请求失败，无数据返回。"}
    else:
        return {"rescode": 200, "resmsg": "请求成功。"}


@shared_task
def doRequestRate(barcode, hash=None):
    """
    查询区域信息  2.3
    barcode,  string 场所编码
    """
    # logger.info(f'[{barcode}] [] 收到异步请求 查询终端列表')
    API = requests_SDK()
    url = "/channel/bar/rate"
    params = {"wb_id": barcode, "code": API.code}
    r = API.do_get(url, params=params)
    if not r:
        logger.warn(f"[query_rate] 查询 区域信息 请求失败")
        return {"rescode": 400, "resmsg": "请求失败，无数据返回。"}
    else:
        return {"rescode": 200, "resmsg": "请求成功。"}


@shared_task
def doRequestRecord(barcode, hash, start_time=None, end_time=None):
    """
    查询消费记录信息  2.4
    barcode,  string 场所编码
    hash=None, string 身份证MD5
    start_time=None, string 开始日期
    end_time=None, string 结束日期
    """
    API = requests_SDK()
    url = "/channel/bar/record"
    params = {"wb_id": barcode, "id_card_hash": hash, "code": API.code}
    if start_time:
        params["start_time"] = start_time
    if end_time:
        params["end_time"] = end_time
    r = API.do_get(url, params=params)
    if not r:
        logger.warn(f"[consumed_record] [{hash}] 查询消费记录 请求失败")
        return {"rescode": 400, "resmsg": "请求失败，无数据返回。"}
    else:
        # 写入 flag 标记
        flag = r.get("data", {}).get("id")
        if flag:
            REDIS.redis_set_key("sysconf", f"FS{flag}", hash, 60 * 1)
        return {"rescode": 200, "resmsg": "请求成功。"}


# ==========================================================================


@shared_task
def doRequestOrder(barcode, hash=None, start_time=None, end_time=None, page=1):
    """
    查询商品订单 3.1
    barcode,  string 场所编码
    hash=None, string 身份证MD5
    start_time=None, string 开始日期
    end_time=None, string 结束日期
    page=1         int 默认第一页
    """
    API = requests_SDK()
    url = "/channel/orders"
    params = {"wb_id": barcode, "code": API.code}
    if hash:
        params["id_card_hash"] = hash
    if start_time:
        params["start_time"] = start_time
    if end_time:
        params["end_time"] = end_time
    if page:
        params["page"] = page
    r = API.do_get(url, params=params)
    return {"rescode": 200, "resmsg": r}


@shared_task
def doRequestRefunds(barcode, hash=None, start_time=None, end_time=None, page=1):
    """
    查询商品退单 3.2
    barcode,  string 场所编码
    hash=None, string 身份证MD5
    start_time=None, string 开始日期
    end_time=None, string 结束日期
    page=1         int 默认第一页
    """
    API = requests_SDK()
    url = "/channel/refunds"
    params = {"wb_id": barcode, "code": API.code}
    if hash:
        params["id_card_hash"] = hash
    if start_time:
        params["start_time"] = start_time
    if end_time:
        params["end_time"] = end_time
    if page:
        params["page"] = page
    r = API.do_get(url, params=params)
    return {"rescode": 200, "resmsg": r}


# ===============================================================
# 接收消息处理
# ===============================================================


@shared_task
def NewdoProgrunList(pid, data):
    """上下机推送消息处理(pid int 场所iD, data dict 数据字典)  1.1-1.2 新 1.4-1.5"""

    def doWriteRedis(pid, dbs):
        """写入缓存"""
        dbs = REDIS.clear_dict(dbs)
        key = f'{pid}--{data.get("computer_name")}'
        REDIS.redis_del("workStatus", key)
        REDIS.redis_hash_set("workStatus", key, dbs)

    def has_Member_Level(wbid, name):
        """查询会员等级"""
        try:
            r = FsygMemberLevel.objects.get(wbcode=wbid, name=name)
        except BaseException as err:
            return 1
        else:
            return r.level

    def to_Count_Week(wbid, cardid):
        """统计每周会员次数"""
        week = date.today().weekday()
        sd = date.today() - timedelta(days=week)
        ed = sd + timedelta(days=7)
        c = FsygProgrunList.objects.filter(
            wb_id=wbid, card_id=cardid, on_day__range=(sd, ed)
        ).aggregate(cishu=Count("id"))
        return c.get("cishu", 0)

    
    """开始处理"""
    guid = data.pop("guid")  # 上机标识
    on_time = timezone.now()  # 上机时间
    # 检查会员信息
    vip, vfaq = FsygMemberList.objects.update_or_create(
        defaults={"lasttime": on_time},
        codeid=data.get("card_id"),
        wb_id=data.get("wb_id"),
    )
    # 检查终端信息
    pcj, pfaq = FsygPchardList.objects.update_or_create(
        defaults={"lasttime": on_time, "status": 1},
        name=data.get("computer_name"),
        pid=pid,
    )
    # 检查上机信息
    run, rfaq = FsygProgrunList.objects.update_or_create(defaults=data, guid=guid)

    if data.get("message_type") == "machine_up":
        # 转移数据 查询上机信息
        event = doRequestProgress(data.get("wb_id"),data.get("card_id"))            
        
        # 记录上机信息
        if rfaq:
            run.on_day = on_time.date()
            run.on_time = on_time
            run.save()
        # 初始化新会员
        if vfaq:
            vip.accummoney = 0  # 累计充值金额(分)
            vip.accumtime = 0  # 累计消费时间(分)
            # vip.overage = "0"  # 卡余额，包含奖励金额
            # vip.reward = "0" # 奖励金额
            vip.currentdate = on_time  # .strftime("%Y-%m-%d %H:%M:%S")
            vip.save()
        # 统计数据
        try:
            # dt = to_Count_Order(data.get("wb_id"), "accumtime", vip.accumtime)
            # dm = to_Count_Order(data.get("wb_id"), "accummoney", vip.accummoney)
            dc = to_Count_Week(data.get("wb_id"), data.get("card_id")) + 1
        except BaseException as err:
            # dt, dm, dc = 0, 0, 1
            dc = 1

        dbs = {
            "status": 1,
            "localhost": pcj.name,  # 终端机号
            "group": pcj.group,
            "ipaddr": pcj.ipaddr,
            "mask": pcj.mask,
            "gateway": pcj.gateway,
            "currenttime": on_time.strftime("%Y-%m-%d %H:%M:%S"),
            "hash": vip.codeid,
            "guid": run.guid,
            "gender": random.choice(range(2)),
            "level": has_Member_Level(pid, vip.level),
            "member": {
                "name": vip.bashname,
                "code": vip.source,
                "hash": vip.codeid,
                "phone": vip.phone,
                "birthday": vip.birthday,
                "gender": vip.gender,
                "openid": vip.openid,
                "registtime": str(vip.currentdate),
                "lasttime": str(vip.lasttime),
                "level": vip.level,
                "overage": vip.overage,
                "reward": vip.reward,
                "integral": vip.integral,
                "accummoney": vip.accummoney,
                "accumtime": vip.accumtime,
                "week": dc,
            },
            "Count": {
                "money": vip.accummoney,
                "domoney": 0,
                "timelong": vip.accumtime,
                "dolong": 0,
                "week": dc,
            },
        }
        doWriteRedis(pid, dbs)
        # 发送消息队列pulsar
        r = doMoniterPulsarScreen(f"{pid}", dbs)
    if data.get("message_type") == "machine_down":
        # 转移数据
        event = UpDownEventFromFsygOff(pid,data)
        evenr = doRequestUser(data.get("wb_id"),data.get("card_id"))
        
        down_time = data.get("down_time")
        dt = datetime.strptime(down_time, "%Y-%m-%d %H:%M:%S")
        dd = dt.date()
        run.down_time = dt
        run.down_day = dd
        run.save()
        # 计算本次上机时长
        ontime = 0 if rfaq else (run.down_time - run.on_time).seconds
        vip.accumtime = vip.accumtime + ontime
        vip.save()

        if rfaq:
            logger.warn(f"[up_down] 未找到上机信息 {guid}")
            # print(f"[up_down] 未找到上机信息 {guid}")
        dbs = {
            "status": 2,  # 终端状态 0维修，1上机中，2待清洁，3待机中
            "localhost": pcj.name,  # 终端机号
            "group": pcj.group,
            "ipaddr": pcj.ipaddr,
            "mask": pcj.mask,
            "gateway": pcj.gateway,
            "currenttime": down_time,
        }
        doWriteRedis(pid, dbs)
        # 创建下机清理
        callme = CreateCallmeClearReqsn(pid, pcj.name)
        # 发送消息队列pulsar
        r = doMoniterPulsarScreen(f"{pid}", dbs)
    logger.info(f"[up_down] 记录完成 {data}")
    # print(f'[新模型] [up_down] [{data.get("message_type")}] 处理上下机记录完成')
    return {"rescode": 200, "resmsg": "ok"}


@shared_task
def DoChangeWorkMessage(data):
    """换机消息推送\n (pid int 场所iD, data dict 数据字典)\n 新1.6"""
    # 转移数据
    event = UpDownEventFromFsygChange(data)
    
    # 查询在线记录
    try:
        r = FsygProgrunList.objects.get(
            message_type="machine_up",
            card_id=data.get("from_card_id"),
            computer_name=data.get("from_computer"),
        )
    except BaseException as err:
        logger.warn(f"[machine_change] [未找到原始上机记录] {data}")
        raise ValueError(f"{err}")
    # 查询场所
    pid = has_pid_from_idcode("凡商编码", r.wb_id)
    if not pid:
        raise ValueError(f"[machine_change] [没有找到该场所] {data}")
    # 插入新算法
    dbs = {
        "status": 2,
        "localhost": data.get("from_computer"),
        "currenttime": timezone.now().strftime("%Y-%m-%d %H:%M:%S"),
    }
    dbs = REDIS.clear_dict(dbs)
    key = f'{pid}--{data.get("from_computer")}'
    REDIS.redis_del("workStatus", key)
    REDIS.redis_hash_set("workStatus", key, dbs)
    # 消息通知
    res = doMoniterPulsarScreen(f"{pid}", dbs)
    # 转移执行
    newdata = {
        "message_type": "machine_up",
        "card_id": r.card_id,
        "computer_name": data.get("to_computer"),
        "guid": r.guid,
        "wb_id": r.wb_id,
        "on_day": r.on_day,
        "on_time": r.on_time,
    }
    res = NewdoProgrunList(pid, newdata)
    logger.info(f"[machine_change] 记录完成 {data}")
    return {"rescode": 200, "resmsg": "ok"}


# @shared_task
# def doProgrunList(pid, data):
#     """上下机推送消息处理(pid int 场所iD, data dict 数据字典)  1.1-1.2"""
#     def to_Count_Order(wbid, key, v):
#         """根据字段计算名次"""
#         m = list(
#             FsygProgrunList.objects.filter(
#                 wb_id=wbid, message_type="machine_up"
#             ).values_list("card_id", flat=True)
#         )
#         r = list(
#             FsygMemberList.objects.filter(wb_id=wbid, codeid__in=m)
#             .values_list(key, flat=True)
#             .order_by(f"-{key}")
#         )
#         for i in range(len(r)):
#             if r[i] == v:
#                 return i + 1
#         return 0

#     def to_Count_Week(wbid, cardid):
#         """统计每周会员次数"""
#         week = date.today().weekday()
#         sd = date.today() - timedelta(days=week)
#         ed = sd + timedelta(days=7)
#         c = FsygProgrunList.objects.filter(
#             wb_id=wbid, card_id=cardid, on_day__range=(sd, ed)
#         ).aggregate(cishu=Count("id"))
#         return c.get("cishu", 0)

#     def has_Member_Level(wbid, name):
#         """查询会员等级"""
#         try:
#             r = FsygMemberLevel.objects.get(wbcode=wbid, name=name)
#         except BaseException as err:
#             return 1
#         else:
#             return r.level

#     """开始正文"""
#     guid = data.pop("guid")

#     # 查询用户会员 标识
#     card_id = data.get("card_id")
#     if not card_id[0] == "-":
#         # 请求 上机信息 查询会员余额、等级
#         msg = doRequestProgress.apply_async((data.get("wb_id"), data.get("card_id")))
#         # 请求 用户信息
#         # msg = doRequestUser.apply_async((data.get("wb_id"), data.get("card_id")))
#         # 请求 区域信息
#         # msg = doRequestRate.apply_async((data.get("wb_id"), data.get("card_id")))
#         # 请求 消费记录
#         # msg = doRequestRecord.apply_async((data.get("wb_id"), data.get("card_id")))
#     # logger.info(f"[up_down] 发送请求 上机信息 区域信息 消费记录 成功")

#     # 检查会员
#     try:
#         v = FsygMemberList.objects.get(
#             wb_id=data.get("wb_id"), codeid=data.get("card_id")
#         )
#     except FsygMemberList.DoesNotExist as err:
#         v = FsygMemberList(
#             overage="0",
#             accumtime=0,
#             accummoney=0,
#             currentdate=timezone.now(),
#             lasttime=timezone.now(),
#             wb_id=data.get("wb_id"),
#             codeid=card_id,
#         )
#         v.save()
#     # logger.info(f"[up_down] 请求 获取会员记录 成功 {v.id}")

#     # 检查终端
#     try:
#         c, faq = FsygPchardList.objects.update_or_create(
#             defaults={"status": 1, "lasttime": timezone.now()},
#             pid=pid,
#             name=data.get("computer_name"),
#         )
#     except BaseException as err:
#         logger.warn(f"[up_down] 请求 检查终端 失败 {err}")
#     else:
#         if faq:
#             # 请求 区域信息
#             msg = doRequestRate.apply_async((data.get("wb_id"), data.get("card_id")))

#     """处理上机数据"""
#     if data.get("message_type") == "machine_up":
#         # 记录会员本次上机时间
#         v.lasttime = timezone.now()
#         v.save()
#         # 创建数据
#         data["on_day"] = timezone.now().date()
#         data["on_time"] = timezone.now()
#         t = v.accumtime or 0
#         m = v.accummoney or 0
#         try:
#             dt = to_Count_Order(data.get("wb_id"), "accumtime", t)
#             dm = to_Count_Order(data.get("wb_id"), "accummoney", m)
#             dc = to_Count_Week(data.get("wb_id"), data.get("card_id"))
#         except BaseException as err:
#             dt, dm, dc = 0, 0, 0
#             logger.info(f"[up_down] [异常监视] {err}")
#         dbs = {
#             "status": 1,
#             "localhost": data.get("computer_name"),
#             "group": c.group,
#             "ipaddr": c.ipaddr,
#             "mask": c.mask,
#             "gateway": c.gateway,
#             "currenttime": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
#             "hash": v.codeid,
#             "guid": guid,
#             "gender": random.choice(range(2)),
#             "level": has_Member_Level(data.get("wb_id"), v.level),
#             "member": {
#                 "name": v.bashname,
#                 "code": v.source,
#                 "hash": v.codeid,
#                 "phone": v.phone,
#                 "birthday": v.birthday,
#                 "gender": v.gender,
#                 "openid": v.openid,
#                 "registtime": str(v.currentdate),
#                 "lasttime": str(v.lasttime),
#                 "level": v.level,
#                 "overage": v.overage,
#                 "reward": v.reward,
#                 "integral": v.integral,
#             },
#             "Count": {
#                 "timelong": t,
#                 "dolong": dt,
#                 "money": m,
#                 "domoney": dm,
#                 "week": dc + 1,
#             },
#         }
#         # logger.info(f"[up_down] 格式化上机数据完场")

#     elif data.get("message_type") == "machine_down":
#         """处理下机数据"""
#         # data["down_day"] = timezone.now().date()
#         data["down_time"] = datetime.strptime(data["down_time"],"%Y-%m-%d %H:%M:%S")
#         data["down_day"] = data["down_time"].date()
#         dbs = {
#             "status": 2,  # 终端状态 0维修，1上机中，2待清洁，3待机中
#             "localhost": data.get("computer_name"),  # 终端机号
#             "group": c.group,
#             "ipaddr": c.ipaddr,
#             "mask": c.mask,
#             "gateway": c.gateway,
#         }
#         # logger.info(f"[up_down] 格式化下机数据完成")
#     else:
#         logger.warn(f"[up_down] 数据无效")
#         return {"rescode": 444, "resmsg": "无效数据"}

#     # 写入缓存数据
#     dbs = REDIS.clear_dict(dbs)
#     key = f'{pid}--{data.get("computer_name")}'
#     REDIS.redis_del("workStatus", key)
#     REDIS.redis_hash_set("workStatus", key, dbs)

#     # REDIS.redis_key_set("workStatus", key, json.dumps(dbs))

#     logger.info(f"[up_down] 写入缓存数据完成")

#     # 写入上下机信息
#     try:
#         sess = data.copy()
#         res, faq = FsygProgrunList.objects.update_or_create(defaults=sess, guid=guid)
#     except BaseException as err:
#         logger.warn(f"[up_down] 写入 在线数据 失败 {err}")

#     if not faq and res.on_time and res.down_time:
#         # if not res.on_time:
#         #     logger.warn(f"[上下机推送] 缺少上机时间")
#         # if not res.down_time:
#         #     logger.warn(f"[上下机推送] 缺少下机时间")
#         # logger.warn(f"[up_down] 测试时间格式 {type(res.on_time)} {res.on_time}")
#         # logger.warn(f"[up_down] 测试时间格式 {type(res.down_time)} {res.down_time}")

#         try:
#             # st = datetime.strptime(res.on_time, "%Y-%m-%d %H:%M:%S")
#             # 累加计算在线时长
#             st = res.on_time
#             et = res.down_time
#             # 计算上机时长
#             wait = (et - st).seconds
#             acctime = v.accumtime or 0
#             v.accumtime = acctime + wait

#             # 计算充值累计 2.4
#             # waim = FsygBuysinfoList.objects.filter(
#             #     wbid=data.get("wb_id"), id_card=card_id, flow_desc="加本金"
#             # ).aggregate(money=Sum("amount"))

#             # 累加计算充值累计 1.3
#             waim = FsygShopintoList.objects.filter(muid=guid, categorize="网费").values(
#                 "price", "count"
#             )
#             wf = [(float(p["price"]) * float(p["count"])) for p in waim]
#             wf = sum(wf)
#             accummoney = v.accummoney or 0
#             v.accummoney = accummoney + int(wf * 100)

#             # 保存数据
#             v.save()
#         except BaseException as err:
#             logger.warn(f"[up_down] [{res.id}] 创建会员统计失败 {err}")

#     if data.get("message_type") == "machine_down":
#         # 创建下机清理
#         callme = CreateCallmeClearReqsn.apply_async((pid, data.get("computer_name")))
#         # logger.info(f"[up_down] 写入会员统计数据完成")

#     # 发送消息队列pulsar
#     r = doMoniterPulsarScreen.apply_async((f"{pid}", dbs))
#     logger.info(f"[up_down] 所有工作处理完成")
#     return {"rescode": 200, "resmsg": "ok"}


@shared_task
def doShopreqsnList(pid, data):
    """商品订、退单消息处理  1.3-1.4"""

    # 查询机器号
    pc = data.get("machine_no", "")
    # 取订单号
    reqsn = data.pop("id")
    # 查询会员身份码
    vip = REDIS.redis_hash_has("workStatus", f"{pid}--{pc}", "hash")
    guid = REDIS.redis_hash_has("workStatus", f"{pid}--{pc}", "guid")
    data["maid"] = vip
    data["guid"] = guid

    # 检测会员
    if vip:
        m, faq = FsygMemberList.objects.update_or_create(
            defaults={"lasttime": timezone.now()},
            wb_id=data.get("wb_id"),
            codeid=vip,
        )

    # 商品流水
    products = data.pop("products")
    for i in products:
        sp = i.pop("product_name")
        s, faq = FsygShopintoList.objects.update_or_create(
            defaults=i,
            wbid=data.get("wb_id"),
            maid=vip,  # 会员
            muid=guid,  # 唯一上机标识
            guid=pc,  # 终端号
            order=reqsn,  # 订单号
            order_id=data.get("order_id"),
            message_type=data.get("message_type"),  # 订单类型
            product_name=sp,
        )
    # logger.info(f"[order] [{reqsn}] [写入商品流水] {products}]")

    # 订单流水
    r, faq = FsygShopreqsnList.objects.update_or_create(defaults=data, reqsn=reqsn)

    # 检查终端
    c, faq = FsygPchardList.objects.update_or_create(
        defaults={"status": 1},
        pid=pid,
        name=pc,
    )

    # 推送商品订单
    if r.order_status in ["1", "2"]:
        dbs = {
            "status": 10,  # 终端状态
            "localhost": pc,  # 终端机号
            "group": c.group,
            "ipaddr": c.ipaddr,
            "mask": c.mask,
            "gateway": c.gateway,
            "order": {"id": r.id, "status": r.order_status},
        }

        # 发送消息队列pulsar
        s = doMoniterPulsarScreen.apply_async((f"{pid}", dbs))
        logger.info(f"[商品订单推送] 写入记录处理完成")

    # 判断网费充值
    if len(products) == 1 and products[0].get("categorize") == "网费":
        return {"rescode": 200, "resmsg": "ok"}

    # 创建呼叫订单
    if r.order_type in ["0", "2"] and r.order_status in ["1", "2"]:
        # if r.order_status == "1" or (r.pay_type == "9" and r.order_status == "0"):
        # logger.info(f'2 准备创建配送呼叫记录 {r.data["id"]}')

        try:
            conf = AllinpayCallmeConf.objects.get(pid=pid, name="配送服务")
        except AllinpayCallmeConf.DoesNotExist as err:
            logger.warn(f"[创建配送] [{pid}] 查询配送服务失败 [{err}]")
            return {"rescode": 430, "resmsg": f"创建配送服务错误 {err}"}

        try:
            grop = FsygPchardList.objects.get(pid=pid, name=pc).group
        except BaseException as err:
            grop = ""

        try:
            staff = AllinpayStaffList.objects.get(basename=r.salesman).id
        except BaseException as err:
            staff = None

        # except BaseException as err:
        #     logger.warning(f"[创建配送] [{pid}] 查询预设数据失败 [{err}]")
        #     return {"rescode": 430, "resmsg": f"创建配送服务错误 {err}"}
        # logger.info(f'3 准备创建配送呼叫记录 {r.data["id"]}')
        if data.get("message_type") == "order":
            dbs = {
                "reqsn": has_Reqsn_Callme(),
                "area": grop,
                "source": pc,
                "current": timezone.now(),
                "rtype": conf.id,
                "status": 0,
                "pid": pid,
                "bem": r.id,  # 商品订单ID
                "target": staff,  # 配送员ID
            }
            c = AllinpayCallmeCreatSerializer(data=dbs)
            c.is_valid()
            if c.errors:
                logger.warn(f"[{c.errors}] 序列化配送订单失败")
                return {"rescode": 420, "resmsg": f"创建商品配送订单错误 {c.errors}"}
            c.save()
            # 派工
            d = c.data
            # res = CALL_TASKS.Dispatch_Reqsn.apply_async(kwargs=AllinpayCallmeCreatSerializer(c).data)
            res = CALL_TASKS.Dispatch_Reqsn(**d)
            # logger.info(f'5 准备创建配送呼叫记录 {res}')
    return {"rescode": 200, "resmsg": "ok"}


# ======================================================================


@shared_task
def doOnlineInfoList(pid, data):
    """上机信息处理  2.1"""
    UpDownEventFromFsygOnn(pid, data)
    return {"rescode": 200, "resmsg": "write onLine is ok"}
    
    
    # 创建数据列表
    wid = data.get("wb_id")
    ulist = data.get("data", {}).get("body", [])
    for i in ulist:
        info = i.copy()
        # guid = info.pop("guid")

        # 写入上机信息
        try:
            name = info.pop("computer_name")
            r, faq = FsygLineinfoList.objects.update_or_create(
                defaults=info, wbid=wid, computer_name=name
            )
        except BaseException as err:
            logger.warn(f"[user_info] 写入上机信息失败 {err}")

        # 写入会员等级与余额
        try:
            fee = info.get("remain_fee") or 0
            m, faq = FsygMemberList.objects.update_or_create(
                defaults={
                    "level": info.get("user_level"),
                    "overage": fee,
                },
                wb_id=wid,
                codeid=info.get("id_card"),
            )
        except BaseException as err:
            logger.warn(f"[user_info] 写入会员信息失败 {err}")

        # 写入终端信息
        try:
            p, faq = FsygPchardList.objects.update_or_create(
                defaults={
                    "mac": info.get("computer_mac"),
                    "ipaddr": info.get("computer_ip"),
                },
                pid=pid,
                name=name,
            )
        except BaseException as err:
            logger.warn(f"[user_info] 写入终端信息失败 {err}")

    return {"rescode": 200, "resmsg": "write onLine is ok"}


@shared_task
def doMemberInfoList(pid, data):
    """用户信息处理 2.2"""
    UpDownEventFromFsygPay(pid, data)
    return {"rescode": 200, "resmsg": "write onLine is ok"}    
    # 创建数据列表
    wid = data.get("wb_id")
    ulist = data.get("data", {}).get("body", [])
    # 更新会员账户信息
    for i in ulist:
        # 写入会员等级与余额
        r, faq = FsygMemberList.objects.update_or_create(
            defaults={
                "level": i["member_level"],  # 会员级别
                "overage": i["overage"],  # 卡余额
                # "reward": i["reward"],  # 奖励金额
                # "integral": i["usable_integral"],  # 可用积分
            },
            wb_id=wid,
            codeid=i["id_card"],
        )
    return {"rescode": 200, "resmsg": "write VIP data is ok"}


@shared_task
def doGroupWorkList(pid, data):
    """区域信息处理 2.3"""
    # 创建数据列表
    wid = data.get("wb_id")
    ulist = data.get("data", {}).get("areas", [])
    # 清空历史数据
    # if ulist:
    #     FsygPchardList.objects.filter(pid=pid).delete()
    # 更新区域数据
    for i in ulist:
        pcs = i.get("computers", "").split(",")
        gtp = i.get("name")
        for j in pcs:
            r, faq = FsygPchardList.objects.update_or_create(
                defaults={"status": 1, "group": gtp}, pid=pid, name=j
            )
    return {"rescode": 200, "resmsg": "write Group Info is ok"}


@shared_task
def doBuyRecordsList(pid, data):
    """消费信息处理 2.4"""
    # 创建数据列表
    wid = data.get("wb_id")
    flag = data.get("flag")
    card_id = REDIS.redis_hash_key("sysconf", f"FS{flag}")
    ulist = data.get("data", {}).get("body", [])
    # 更新数据
    for i in ulist:
        # 其他唯一主键标识
        mhash = hash_md5(f"{i}")
        r, faq = FsygBuysinfoList.objects.update_or_create(
            defaults=i, wbid=wid, card_id=card_id, mhash=mhash
        )

    return {"rescode": 200, "resmsg": "write buy log is ok"}
